package com.lin.fengxiu.manager.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

/**
 * @ClassName ConsumerSchedule
 * @Description TODO 消费者
 * @Author wushaopei
 * @Date 2021/4/17 10:55
 * @Version 1.0
 */
@Component
public class ConsumerSchedule implements CommandLineRunner{

    @Value("${rocketmq.consumer.consumer-group}")
    private String consumerGroup;

    @Value("${rocketmq.namesrv-addr}")
    private String namesrvAddr;

    public void messageListener() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.consumerGroup);

        // 消费者消费的地址 ip+port
        consumer.setNamesrvAddr(this.namesrvAddr);
        // 消费者订阅的主题，topic代表主题名字、* 代表所有消息
        consumer.subscribe("TopicTest","*");
        // 一次消费的消息数量
        consumer.setConsumeMessageBatchMaxSize(1);
        // 注册监听器，读取生产者发布的消息
        // message 、context 相当于 callback 结果
        consumer.registerMessageListener((MessageListenerConcurrently)(messages, context)->{
            for (Message message: messages){
                System.out.println("消息：" + new String(message.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }

    @Override
    public void run(String... args) throws Exception {
        this.messageListener();
    }
}
